import os, { hostname } from 'os';
import util from 'util';
import { injectable } from 'inversify';
import { createHandyClient, IHandyRedis } from 'handy-redis';
import { ISyncService } from '../api/sync';
import { makeLabelLogger, lerror } from '../../util/logger';
import { sleep } from '../../util/tools';
import { EventEmitter } from 'events';

let { llog } = makeLabelLogger('sync');
let localChannelName = 'mq-user';
let remoteChannelName = 'mq-sync';

interface IHandyRedisEx extends IHandyRedis {
	ConnectWaiter: (from?: string) => Promise<{}>;
	config: any;
	name: string;
	ready: boolean;
}

@injectable()
export class SyncService implements ISyncService {
	started: boolean;
	public hostname: string;
	remoteRedisCommon: IHandyRedisEx; // 用来给远程发送命令
	remoteRedisWatchingStream: IHandyRedisEx; // 用来阻塞读取远程流事件

	localRedisCommon: IHandyRedisEx;
	localRedisWatching: IHandyRedisEx;

	remoteMsgEvent = new EventEmitter();

	public async postMessage(message: any): Promise<boolean> {
		if (!this.hostname) {
			this.hostname = await this.localRedisCommon.get('config:hostname');
			// llog('loadHostname3', this.hostname);
		}
		if (!this.hostname) {
			llog('还未设置hostname，不能发消息！');
			return;
		}

		let msg = message;
		msg.srcHost = this.hostname;
		let str = JSON.stringify(message);
		llog('post', str);
		this.localRedisCommon.rpush(localChannelName, str);
		return true;
	}
	public onRemoteMessage(type: string, handler: (msg: any) => void): void {
		this.remoteMsgEvent.on(type, handler);
	}

	public isOwnAction(action): boolean {
		return action.srcHost == this.hostname;
	}

	public start(localRedisConfig, syncRedisConfig) {
		if (this.started) return;
		this.started = true;

		// 创建远程连接
		syncRedisConfig = Object.assign(syncRedisConfig, { enable_offline_queue: false });
		// syncRedisConfig = Object.assign(syncRedisConfig, { enable_offline_queue: true });
		this.remoteRedisCommon = this.createRedisConnection({ ...syncRedisConfig, name: 'remoteRedisCommon' });
		this.remoteRedisWatchingStream = this.createRedisConnection({ ...syncRedisConfig, name: 'remoteRedisWatchingStream' });

		// 创建本地连接，一个用来发，一个用来阻塞读
		this.localRedisCommon = this.createRedisConnection({ ...localRedisConfig, name: 'localRedisCommon' });
		this.localRedisWatching = this.createRedisConnection({ ...localRedisConfig, name: 'localRedisWatching' });

		llog('sync starting...');
		llog('local', localRedisConfig);
		llog('remote', syncRedisConfig);

		this.loadHostname();

		this.WatchQueue();
		this.WatchStream();
	}

	async loadHostname() {
		await this.localRedisCommon.ConnectWaiter('loadHostname');
		this.hostname = await this.localRedisCommon.get('config:hostname');
		llog('loadHostname', this.hostname);
	}

	createRedisConnection(config) {
		let hredis = createHandyClient(config) as IHandyRedisEx;
		let url = `${config.host}:${config.port}`;
		hredis.redis.on('error', () => {
			llog(`${config.name} redis disconnect: ${url}`);
			hredis.ready = false;
		});
		hredis.redis.on('ready', () => {
			// llog(`${config.name} redis ready: ${url}`);
			hredis.ready = true;
		});
		hredis.config = config;
		hredis.ready = false;
		hredis.ConnectWaiter = this.createRedisWaiter(hredis);
		return hredis;
	}
	createRedisWaiter(hredis: IHandyRedisEx) {
		let config = hredis.config;
		let url = `${config.host}:${config.port}`;
		let createWaitConnectPromise = (from?: string) => {
			let fillWhenConnect: () => void;
			// llog(`${config.name} redis wait-->: ${url}, waitfrom: ${from || ''}`, hredis.redis.connected, hredis.redis.connection_id, hredis.ready);
			hredis.redis.once('connect', () => {
				// llog(`${config.name} redis connect: ${url}, waitfrom: ${from || ''}`);
			});
			hredis.redis.once('ready', () => {
				llog(`${config.name} redis ready: ${url}, waitfrom: ${from || ''}`);
				hredis.ready = true;
				if (fillWhenConnect) {
					fillWhenConnect();
					fillWhenConnect = null;
				}
			});
			return new Promise(fill => {
				if (hredis.ready) {
					llog(`${config.name} redis already: ${url}, waitfrom: ${from || ''}`);
					fill();
				} else fillWhenConnect = fill;
			});
		};
		return createWaitConnectPromise;
	}

	async WatchQueue() {
		await this.localRedisWatching.ConnectWaiter('watchqueue begin');
		llog('watching local queue ');
		while (true) {
			let msg;

			try {
				[, msg] = await this.localRedisWatching.blpop([localChannelName], 0);
				// llog('get msg', msg);
			} catch (e) {
				//
				if (!this.localRedisWatching.redis.connected) {
					await this.localRedisWatching.ConnectWaiter('watchqueue break');
				}
				continue;
			}

			try {
				// llog('prepare send msg ok');
				// 先塞回去
				this.localRedisWatching.lpush(localChannelName, msg);
				// 发送
				await this.remoteRedisCommon.rpush(remoteChannelName, msg);
				// 发送成功了再删除，确保未发送成功时，消息不丢
				this.localRedisWatching.lpop(localChannelName);
				// 发送
			} catch (e) {
				llog('push msg fail', msg);
				if (!this.remoteRedisCommon.redis.connected) {
					llog('remote redis disconnect, wait for connect');
					// this.localRedisWatching.lpush(localChannelName, JSON.stringify({ type: 'mark', info: 'offline' }));
					await this.remoteRedisCommon.ConnectWaiter('push msg break');
					llog('remote redis connect again');
				}
			}
		}
	}

	async WatchStream() {
		// llog('watching stream---------->>');
		await this.localRedisCommon.ConnectWaiter('watchstream begin');
		await this.remoteRedisWatchingStream.ConnectWaiter('watchstream begin');
		// llog('watching stream<<----------');
		let readFrom = (await this.localRedisCommon.get('last_sync')) || '$';
		while (true) {
			llog(`watching remote stream,  from:${readFrom}`);

			try {
				let msg = await new Promise((fill, reject) => {
					(<any>this.remoteRedisWatchingStream.redis).xread('block', 60 * 1000, 'streams', 'ub', readFrom, (err, reply) => {
						if (err) reject(err);
						else fill(reply);
					});
				});
				if (!msg) {
					// 超时了，没读到东西
					continue;
				}
				let [t, item] = msg[0][1][0];
				let [, type, , body] = item;
				llog(`get remote msg, time:${t}, type:${type}`);
				// console.log(util.inspect(msg, false, 10, true));

				// 发出事件，由注册者处理
				try {
					this.remoteMsgEvent.emit(type, JSON.parse(body));
				} catch (e) {
					lerror('dispatch remote msg fail', e);
				}

				// 记录下一个点
				readFrom = msg[0][1][0][0];
				this.localRedisCommon.set('last_sync', readFrom);
			} catch (e) {
				llog('get remote msg fail ', e);
				if (!this.remoteRedisWatchingStream.redis.connected) {
					await this.remoteRedisWatchingStream.ConnectWaiter('watchstream break');
				}
			}
		}
	}
}
